package cn.ddiancan.xddcloud.psc.mq;

import java.util.List;

import cn.ddiancan.xddcloud.psc.mq.config.XddCloudMqProperties;
import cn.ddiancan.xddcloud.psc.mq.constants.ListenerType;
import cn.ddiancan.xddcloud.psc.mq.factory.XddcloudProducerFactory;
import cn.ddiancan.xddcloud.psc.mq.service.XddcloudMessageListener;
import cn.ddiancan.xddcloud.psc.mq.service.XddcloudOrderlyMessageListener;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.util.StringUtils;

@Slf4j
@AutoConfiguration
@EnableConfigurationProperties(XddCloudMqProperties.class)
@ConditionalOnProperty(prefix = "xddcloud.mq", name = "enabled", matchIfMissing = true)
public class XddcloudMqAutoConfiguration implements InitializingBean {

    @Autowired
    private XddCloudMqProperties xddCloudMqProperties;

    @Autowired
    private List<IConfigHandler> configHandlers;

    @Override
    public void afterPropertiesSet() {
        initProducer();
        initConsumer();
    }

    public void initConsumer() {
        for (IConfigHandler configHandler : configHandlers) {
            AbstractMessageHandler messageHandler = (AbstractMessageHandler) configHandler;
            if (messageHandler.isEnableCustomer()) {
                String topic = messageHandler.getTopic();
                try {
                    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(topic);
                    consumer.setNamesrvAddr(xddCloudMqProperties.getNamesrvAddr());
                    consumer.setMessageModel(messageHandler.getMessageModel());
                    consumer.subscribe(messageHandler.getTopic(),
                        StringUtils.hasText(messageHandler.getTag()) ? messageHandler.getTag() : "*");
                    if (ListenerType.CONCURRENTLY.equals(messageHandler.listenerType())) {
                        consumer.registerMessageListener(new XddcloudMessageListener(topic));
                    } else {
                        consumer.registerMessageListener(new XddcloudOrderlyMessageListener());
                    }

                    consumer.start();
                    log.info("启动消费者成功，topic:{}", topic);
                } catch (Exception e) {
                    log.error("启动消费者错误：{},topic:{}", e.getMessage(), topic);
                }
            }
        }
    }

    public void initProducer() {

        for (IConfigHandler configHandler : configHandlers) {
            if (configHandler.isEnableProducer()) {
                String topic = configHandler.getTopic();
                DefaultMQProducer producer = new DefaultMQProducer(topic);
                producer.setNamesrvAddr(xddCloudMqProperties.getNamesrvAddr());
                try {
                    producer.start();
                    log.info("启动生产者成功，topic:{}", topic);
                    XddcloudProducerFactory.putProducer(topic, producer);
                } catch (Exception e) {
                    log.error("启动生产者错误：{},topic:{}", e.getMessage(), topic);
                }
            }
        }

    }
}
